package com.stillcoolme.framework.util;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;

public class Mean {

    public static class Map extends Mapper<LongWritable, Text, Text, DoubleWritable>{
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] record = value.toString().split(",");
            System.out.println(record[8]);
            //判断数据是否正确
            if (record != null && record.length == 12) {
                String gender = record[5];
                System.out.println(gender);
                Double score = Double.parseDouble(record[8]);
                System.out.println(score);

                context.write(new Text(gender), new DoubleWritable(score));
            }
        }
    }

    // 在数据达到reducer之前，MapReduce框架已经按照key值对这些数据按键排序了，就是shuffle()
    // 如果key为封装的int为IntWritable类型，那么MapReduce按照数字大小对key排序
    // 如果Key为封装String的Text类型，那么MapReduce将按照数据字典顺序对字符排序
    // 所以一般在map中把要排序的字段使用IntWritable类型，作为key，不排序的字段作为value
    public static class Reduce extends Reducer<Text, DoubleWritable, Text, DoubleWritable>{
        @Override
       protected void reduce(Text key, Iterable<DoubleWritable> values, Reducer<Text, DoubleWritable, Text, DoubleWritable>.Context context) throws IOException, InterruptedException {
            double count = 0.0;
            int n = 0;
            //获取到迭代器开始遍历数据
            for (DoubleWritable value : values){
                count += value.get();
                n++;
            }
            //写出数据
            context.write(key, new DoubleWritable(count/n));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration(true);
        configuration.set("fs.defaultFS","hdfs://hadoop102:8020");
        //本地模式运行
        configuration.set("mapreduce.framework.name", "local");
        //创建Job
        Job job = Job.getInstance(configuration);
        job.setJobName("Mean"+ new SimpleDateFormat("yyyyMMdd-HHmmss-SSS").format(new Date()));
        job.setJarByClass(Mean.class);


        job.setNumReduceTasks(1);

        FileInputFormat.setInputPaths(job, new Path("/score/data/2020jsj.csv"));
        FileOutputFormat.setOutputPath(job, new Path("/score/result/" + job.getJobName()));


        job.setMapperClass(Map.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);

        job.setReducerClass(Reduce.class);

        job.waitForCompletion(true);

    }
}
